package com.qudian.pay.mq.consumer;

import java.util.List;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.util.StopWatch;
import com.alibaba.fastjson.JSON;
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.qudian.pay.core.utils.JdkSerializer;
import com.qudian.pay.mq.MQException;
import com.qudian.pay.mq.ReceiveMessage;
import com.qudian.pay.mq.SendMessage;
import com.qudian.pay.mq.TopicTagAssemble;
import com.qudian.pay.mq.enums.MQConsumeStatus;
import com.qudian.pay.mq.enums.MQRuntimeFailStatus;

/**
 * @Description: 同步消费message抽象类.同一队列的消息会并行消费,不保证消息在同一队列严格有序消费
 * doConsume在获取消息的过程中同步调用,如果业务方法执行过长,
 * 会导致rocketmq阻塞到timeout(pullInterval的时间间隔)并触发自动重发 
 * 好处: 编程简单 
 * 坏处: 阻塞broker,若导致重发,调用端需要写代码处理重发的情况
 * (尤其是消息第一次到达后,业务尚未处理完毕,重发的消息又到了导致业务代码多次操作数据库的情况)
 * TODO 将客户端获取msg的模式改为异步 AbstractMQAsyncConsumerConcurrently
 * 好处: 隔离应用与rocketmq
 * consumer拉取线程,不阻塞broker,不需要设置pullInterval值(保持默认0就OK) 
 * 坏处: 编程复杂度提高
 * @author gaoxiang
 * @date 2016年5月23日 下午1:21:18
 */ 
public abstract class AbstractMQConsumer implements MessageListenerConcurrently {
	
	protected final static Logger logger = LoggerFactory.getLogger(AbstractMQConsumer.class);
	
	/**
	 * @Fields DELAY_LEVEL : 消费失败时，下次消费延时时间
	 */
	protected int DELAY_LEVEL = 0;
	
	/**
	 * @Fields namesrvAddr : nameServer IP:Port
	 */
	protected String namesrvAddr;
	
	protected String instanceName;
	
	protected String group;
	
	protected String topic;
	
	/**
	 * @Fields tagExpress : broker依据此表达式进行过滤。目前只支持或运算,例如: "tag1 || tag2 || tag3" 如果subExpression等于null或者*，则表示全部订阅
	 */
	protected String tagExpress;
	
	/**
	 * @Fields topicsACL : 允许接收的topics,多个使用逗号隔开
	 */
	protected String topicsACL;
	
	/**
	 * @Fields tagsACL : 允许接收的tags,多个使用逗号隔开
	 */
	protected String tagsACL;
	
	/**
	 * @Fields needCheckTopics : 是否对可以接收的topic进行检查的标示
	 */
	protected boolean needCheckTopics = true;
	
	/**
	 * 是否对可以接收的tag进行检查的标示
	 */
	protected boolean needCheckTags = true;
	
	/**
	 * @Fields consumeTimesThreshold : 消费次数阀值
	 */
	protected int consumeTimesThreshold = 15;
	
	protected DefaultMQPushConsumer consumer;
	
	/**
	 * @Fields consumeThreadMin : 消费消息线程，最小数目
	 */
	protected int consumeThreadMin = 20;
	
	/**
	 * @Fields consumeThreadMax : 消费消息线程，最大数目
	 */
	protected int consumeThreadMax = 100;
	
	/**
	 * @Fields pullIntervalInMillis : 拉消息间隔,设置为10秒(设置太长会降低系统TPS)
	 */
	protected String pullIntervalInMillis;
	
	/**
	 * @Description: consumer销毁
	 * @author gaoxiang
	 * @date 2017年2月6日 下午6:28:05
	 */
	public void destroy() {
		if (consumer != null) {
			consumer.shutdown();
			logger.info("Consumer Shutdown Success.topic={" + getTopic() + "},tagExpress={" + getTagExpress() + "}");
		}
	}
	
	/**
	 * @Description: 具体业务处理 针对不同业务系统进行相应的个性化业务后续处理
	 * @author gaoxiang
	 * @date 2017年2月6日 下午6:28:20
	 */
	public abstract MQConsumeStatus doConsume(ReceiveMessage msg) throws MQException;
	
	/**
	 * @Description: doConsume如果出现异常,调用handleMQException方法处理
	 * (比如将一些业务数据持久化至数据库,再通过其他方式处理)
	 * 此方法不允许抛出异常,需要自行处理掉
	 * @param msg
	 * @return void 返回类型
	 * @author gaoxiang
	 * @date 2016年6月6日 上午10:40:32
	 */
	public abstract void handleMQException(ReceiveMessage msg);
	
	/**
	 * @Description: mq消费者启动配置检测
	 * @author gaoxiang
	 * @date 2017年2月6日 下午6:29:39
	 */
	public void checkPropertiesAndSetEnv() {
		TopicTagAssemble topicAndTag = getTopicAndTag();
		if (null != topicAndTag) {
			topic = topicAndTag.getTopic();
			tagExpress = topicAndTag.getTag();
			group = "S_" + topic + "_" + tagExpress;
		}
		if (StringUtils.isEmpty(namesrvAddr) || StringUtils.isEmpty(instanceName) || StringUtils.isEmpty(getGroup())) {
			logger.error("未设置 namesrv or instanceName or group");
			throw new MQException(MQRuntimeFailStatus.MQ_INIT_PARAM_FAILURE);
		}
		// setting nameserver environment variable
		if (StringUtils.isEmpty(System.getProperty("rocketmq.namesrv.addr")) && StringUtils.isNotEmpty(namesrvAddr)) {
			System.setProperty("rocketmq.namesrv.addr", namesrvAddr);
		}
		if (StringUtils.isEmpty(topic) || StringUtils.isEmpty(tagExpress)) {
			logger.error("Topic or tag 不可为空");
			throw new MQException(MQRuntimeFailStatus.MQ_INIT_PARAM_FAILURE);
		}
		// 检查topic及tag是否在ACL列表中
		if (doCheckTopic(topic) || doCheckTag(tagExpress)) {
			logger.error("没有权限.Topic={" + topic + "},tagExpress={"+ tagExpress + "},topicsACL={" + topicsACL + "},tagsACL={" + tagsACL + "}");
			throw new MQException(MQRuntimeFailStatus.MQ_PUBLISH_PERMISSION_FAILURE);
		}
	}
	
	public boolean doCheckTopic(String topic) {
		if (needCheckTopics && StringUtils.isNotEmpty(topicsACL)) {
			if (topicsACL.contains(topic)) {
				return true;
			}
			logger.error("该话题未在允许列表里:topic={" + topic + "},允许列表是{" + topicsACL + "}");
			return false;
		}
		return false;
	}
	
	public boolean doCheckTag(String tag) {
		if (needCheckTags && StringUtils.isNotEmpty(tagsACL)) {
			if (topicsACL.contains(tag)) {
				return true;
			}
			logger.error("This tag is not in allowed tag list:tag={" + tag + "},allowed tag list={" + tagsACL + "}");
		}
		return false;
	}
	
	protected String getGroup() {
		if (StringUtils.isEmpty(group)) {
			group = "S_" + topic + "_" + tagExpress;
		}
		return group;
	}
	
	/**
	 * 返回topic及tag(tag必须是属于这个topic下的) 如果想通过程序设置topic及tag属性，可以重载该方法
	 * (目前的规则是程序设置的优先级大于在配置文件中设置的)
	 */
	public TopicTagAssemble getTopicAndTag() {
		return null;
	}
	
	@Override
	public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
		if (null == msgs || msgs.size() != 1) {
			logger.error("消费失败，消息长度不为1:Topic={" + topic + "},tags={" + tagExpress + "},group={" + getGroup() + "},msgs={" + JSON.toJSONString(msgs)
					+ "},msgs.size={" + (msgs == null ? 0 : msgs.size()) + "}");
			return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
		}
		MessageExt msgExt = msgs.get(0);
		if (msgExt.getReconsumeTimes() >= consumeTimesThreshold) {
			logger.info("(" + msgExt.getMsgId() + ")" + "topic:" + msgExt.getTopic() + " tags:" + msgExt.getTags() + " 消费次数超限:"+consumeTimesThreshold);
			return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
		}
		StopWatch stopWatch = new StopWatch();
		ReceiveMessage receiveMessage = new ReceiveMessage();
		BeanUtils.copyProperties(msgExt, receiveMessage);
		try {
			SendMessage sendMessage = JdkSerializer.deSerialize(msgExt.getBody());
			receiveMessage.setBizValue(sendMessage.getBizValue());
			receiveMessage.setInstanceName(sendMessage.getInstanceName());
		} catch (Exception e) {
			throw new MQException(MQRuntimeFailStatus.MQ_SERIALIZE_FAILURE, e);
		}
		try {
			stopWatch.start();
			MQConsumeStatus result = doConsume(receiveMessage);
			stopWatch.stop();
			logger.info("消费成功，消费时长:{" + stopWatch.toString() + "},msgDetail:{" + msgExt.toString() + "},group={" + getGroup() + "}");
			if (result.equals(MQConsumeStatus.SUCCESS) || result.equals(MQConsumeStatus.HANDLED)) {
				return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
			} else {
				if (DELAY_LEVEL > 0) context.setDelayLevelWhenNextConsume(DELAY_LEVEL);
				return ConsumeConcurrentlyStatus.RECONSUME_LATER;
			}
		} catch (MQException e) {
			handleMQException(receiveMessage);
			stopWatch.stop();
			logger.error("消费成功，消费时长:{" + stopWatch.toString() + "},msg={" + msgExt.toString() + "},group={" + getGroup() + "}", e);
			return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
		} finally {
			if (stopWatch.isRunning()) stopWatch.stop();
			stopWatch = null;
		}
	}
	
	public void setDelayLevel(int level) {
		this.DELAY_LEVEL = level;
	}
	
	public String getNamesrvAddr() {
		return namesrvAddr;
	}
	
	public void setNamesrvAddr(String namesrvAddr) {
		this.namesrvAddr = namesrvAddr;
	}
	
	public String getTopic() {
		return topic;
	}
	
	public void setTopic(String topic) {
		this.topic = topic;
	}
	
	public String getTagExpress() {
		return tagExpress;
	}
	
	public void setTagExpress(String tagExpress) {
		this.tagExpress = tagExpress;
	}
	
	public String getInstanceName() {
		return instanceName;
	}
	
	public void setInstanceName(String instanceName) {
		this.instanceName = instanceName;
	}
	
	public String getTopicsACL() {
		return topicsACL;
	}
	
	public void setTopicsACL(String topicsACL) {
		this.topicsACL = topicsACL;
	}
	
	public String getTagsACL() {
		return tagsACL;
	}
	
	public void setTagsACL(String tagsACL) {
		this.tagsACL = tagsACL;
	}
	
	public String getPullIntervalInMillis() {
		return pullIntervalInMillis;
	}
	
	public void setPullIntervalInMillis(String pullIntervalInMillis) {
		this.pullIntervalInMillis = pullIntervalInMillis;
	}
	
	public int getConsumeThreadMin() {
		return consumeThreadMin;
	}
	
	public void setConsumeThreadMin(int consumeThreadMin) {
		this.consumeThreadMin = consumeThreadMin;
	}
	
	public int getConsumeThreadMax() {
		return consumeThreadMax;
	}
	
	public void setConsumeThreadMax(int consumeThreadMax) {
		this.consumeThreadMax = consumeThreadMax;
	}
	
	public int getConsumeTimesThreshold() {
		return consumeTimesThreshold;
	}
	
	public void setConsumeTimesThreshold(int consumeTimesThreshold) {
		this.consumeTimesThreshold = consumeTimesThreshold;
	}
}
